package com.atguigu.ting.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.ting.realtime.app.BaseApp;
import com.atguigu.ting.realtime.bean.AlbumOrderBean;
import com.atguigu.ting.realtime.common.Constant;
import com.atguigu.ting.realtime.funtion.DimAsyncFunction;
import com.atguigu.ting.realtime.util.AtguiguUtil;
import com.atguigu.ting.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @Author lzc
 * @Date 2023/4/14 19:17
 */
public class DwsAlbumOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsAlbumOrderWindow().init(
            40006,
            2,
            "DwsAlbumOrderWindow",
            Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       DataStreamSource<String> stream) {
        //1. 解析成 pojo 类型
        SingleOutputStreamOperator<AlbumOrderBean> beanStream = parseToPojo(stream);
        // 2. 按照订单详情 id 去重
        beanStream = distinctByOrderDetailId(beanStream);
        // 3. 分组开窗聚合
        SingleOutputStreamOperator<AlbumOrderBean> beanStreamWithoutDims = windowAndAgg(beanStream);
        // 4. 补充维度信息
        SingleOutputStreamOperator<AlbumOrderBean> beanStreamWithDims = joinDim(beanStreamWithoutDims);
        // 5. 写出到 clickhouse 中
        writeToClickHouse(beanStreamWithDims);
    }
    
    private void writeToClickHouse(SingleOutputStreamOperator<AlbumOrderBean> beanStreamWithDims) {
        beanStreamWithDims.addSink(FlinkSinkUtil.getClickHouseSink("dws_album_order_window", AlbumOrderBean.class));
    }
    
    private SingleOutputStreamOperator<AlbumOrderBean> joinDim(
        SingleOutputStreamOperator<AlbumOrderBean> beanStreamWithoutDims) {
        SingleOutputStreamOperator<AlbumOrderBean> albumStream = AsyncDataStream.unorderedWait(
            beanStreamWithoutDims,
            new DimAsyncFunction<AlbumOrderBean>() {
                @Override
                public void addDim(AlbumOrderBean bean,
                                   JSONObject dim) {
                    bean.setAlbumName(dim.getString("ALBUM_TITLE"));
                    bean.setCategory1Id(dim.getString("CATEGORY1_ID"));
                    bean.setCategory2Id(dim.getString("CATEGORY2_ID"));
                    if (dim.getString("CATEGORY3_ID") == null) {
                        System.out.println(dim);
                    } else {
                        System.out.println("dim: " + dim);
                    }
                    bean.setCategory3Id(dim.getString("CATEGORY3_ID"));
                }
                
                @Override
                public String getId(AlbumOrderBean bean) {
                    return bean.getAlbumId();
                }
                
                @Override
                public String getTable() {
                    return "dim_album_info";
                }
            },
            60,
            TimeUnit.SECONDS
        );
        SingleOutputStreamOperator<AlbumOrderBean> c3Stream = AsyncDataStream.unorderedWait(
            albumStream,
            new DimAsyncFunction<AlbumOrderBean>() {
                @Override
                public void addDim(AlbumOrderBean bean,
                                   JSONObject dim) {
                    bean.setCategory3Name(dim.getString("NAME"));
                }
                
                @Override
                public String getId(AlbumOrderBean bean) {
                    return bean.getCategory3Id();
                }
                
                @Override
                public String getTable() {
                    return "dim_base_category3";
                }
            },
            60,
            TimeUnit.SECONDS
        );
        SingleOutputStreamOperator<AlbumOrderBean> c2Stream = AsyncDataStream.unorderedWait(
            c3Stream,
            new DimAsyncFunction<AlbumOrderBean>() {
                @Override
                public void addDim(AlbumOrderBean bean,
                                   JSONObject dim) {
                    bean.setCategory2Name(dim.getString("NAME"));
                }
                
                @Override
                public String getId(AlbumOrderBean bean) {
                    return bean.getCategory2Id();
                }
                
                @Override
                public String getTable() {
                    return "dim_base_category2";
                }
            },
            60,
            TimeUnit.SECONDS
        );
        return AsyncDataStream.unorderedWait(
            c2Stream,
            new DimAsyncFunction<AlbumOrderBean>() {
                @Override
                public void addDim(AlbumOrderBean bean,
                                   JSONObject dim) {
                    bean.setCategory1Name(dim.getString("NAME"));
                }
                
                @Override
                public String getId(AlbumOrderBean bean) {
                    return bean.getCategory1Id();
                }
                
                @Override
                public String getTable() {
                    return "dim_base_category1";
                }
            },
            60,
            TimeUnit.SECONDS
        );
    }
    
    
    private SingleOutputStreamOperator<AlbumOrderBean> windowAndAgg(
        SingleOutputStreamOperator<AlbumOrderBean> beanStream) {
        return beanStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<AlbumOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
                    .withIdleness(Duration.ofSeconds(60))
            )
            .keyBy(AlbumOrderBean::getAlbumId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<AlbumOrderBean>() {
                    @Override
                    public AlbumOrderBean reduce(AlbumOrderBean value1,
                                                 AlbumOrderBean value2) throws Exception {
                        value1.setSplitOriginalAmount(value1.getSplitOriginalAmount().add(value2.getSplitOriginalAmount()));
                        value1.setSplitDerateAmount(value1.getSplitDerateAmount().add(value2.getSplitDerateAmount()));
                        value1.setSplitOrderAmount(value1.getSplitOrderAmount().add(value2.getSplitOrderAmount()));
                        
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        value1.getUidSet().addAll(value2.getUidSet());
                        
                        return value1;
                    }
                },
                new ProcessWindowFunction<AlbumOrderBean, AlbumOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(String skuId,
                                        Context ctx,
                                        Iterable<AlbumOrderBean> elements,
                                        Collector<AlbumOrderBean> out) throws Exception {
                        AlbumOrderBean bean = elements.iterator().next();
                        
                        bean.setStt(AtguiguUtil.tsToDateTime(ctx.window().getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(ctx.window().getEnd()));
                        
                        bean.setOrderCt((long) bean.getOrderIdSet().size());
                        bean.setOrderUserCt((long) bean.getUidSet().size());
                        
                        bean.setTs(System.currentTimeMillis());
                        
                        out.collect(bean);
                    }
                }
            );
    }
    
    
    private SingleOutputStreamOperator<AlbumOrderBean> distinctByOrderDetailId(
        SingleOutputStreamOperator<AlbumOrderBean> beanStream) {
        return beanStream
            .keyBy(AlbumOrderBean::getOrderDetailId)
            .process(new KeyedProcessFunction<String, AlbumOrderBean, AlbumOrderBean>() {
                
                private ValueState<AlbumOrderBean> state;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    ValueStateDescriptor<AlbumOrderBean> desc =
                        new ValueStateDescriptor<>("beanState", AlbumOrderBean.class);
                    StateTtlConfig ttlConfig = new StateTtlConfig.Builder(org.apache.flink.api.common.time.Time.seconds(10))
                        .updateTtlOnCreateAndWrite()
                        .build();
                    
                    desc.enableTimeToLive(ttlConfig);
                    state = getRuntimeContext().getState(desc);
                }
                
                @Override
                public void processElement(AlbumOrderBean currentBean,
                                           Context ctx,
                                           Collector<AlbumOrderBean> out) throws Exception {
                    AlbumOrderBean lastBean = state.value();
                    if (lastBean == null) { // 当前详情 id 第一条数据
                        out.collect(currentBean); // 直接输出
                    } else { // 不是第一条
                        // 用新的数据, 减去状态中的数据, 输出
                        lastBean.setSplitOriginalAmount(currentBean.getSplitOriginalAmount().subtract(lastBean.getSplitOriginalAmount()));
                        lastBean.setSplitDerateAmount(currentBean.getSplitDerateAmount().subtract(lastBean.getSplitDerateAmount()));
                        lastBean.setSplitOrderAmount(currentBean.getSplitOrderAmount().subtract(lastBean.getSplitOrderAmount()));
                        
                        out.collect(lastBean);
                    }
                    state.update(currentBean);  // 更新状态
                }
            });
    }
    
    
    private SingleOutputStreamOperator<AlbumOrderBean> parseToPojo(DataStreamSource<String> stream) {
        return stream
            .process(new ProcessFunction<String, AlbumOrderBean>() {
                @Override
                public void processElement(String value,
                                           Context ctx,
                                           Collector<AlbumOrderBean> out) throws Exception {
                    JSONObject obj = JSON.parseObject(value);
                    
                    String itemType = obj.getString("item_type");
                    if ("1702".equals(itemType)) { // 专辑下单
                        AlbumOrderBean bean = AlbumOrderBean.builder()
                            .orderDetailId(obj.getString("order_detail_id"))
                            .albumId(obj.getString("item_id"))
                            .orderIdSet(new HashSet<>(Collections.singletonList(obj.getString("order_id"))))
                            .uidSet(new HashSet<>(Collections.singletonList(obj.getString("user_id"))))
                            .splitOriginalAmount(obj.getBigDecimal("split_original_amount"))
                            .splitDerateAmount(obj.getBigDecimal("split_derate_amount"))
                            .splitOrderAmount(obj.getBigDecimal("split_order_amount"))
                            .ts(obj.getLong("ts") * 1000)
                            .build();
                        
                        out.collect(bean);
                    }
                }
            });
    }
}
